package com.one.api.common.msg.sender;

import com.one.api.common.constant.BeanNameConstant;
import com.one.api.common.msg.constant.MsgConstant;
import com.one.api.common.msg.entity.Msg;
import com.one.api.common.msg.entity.StringMsg;
import net.sf.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component(BeanNameConstant.kafkaSender)
public class KafkaSender {
    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendEmail(Msg msg){
        send(MsgConstant.emailKafkaTopic, msg);
//        kafkaTemplate.send(MsgConstant.emailKafkaTopic, JSONObject.fromObject(msg).toString());
    }
    public void sendString(StringMsg msg){
        send(MsgConstant.emailKafkaTopic, msg);
    }

    public void send(String topic, Object msg){
        String msgStr=JSONObject.fromObject(msg).toString();
        logger.debug("to send kafka msg, topic=["+topic+"], msg=["+msgStr+"]");
        kafkaTemplate.send(topic, msgStr);
    }

    public void send(String topic, String msg){
        kafkaTemplate.send(topic, msg);
    }

}
